/*
 *   Servlet for surfacing syndicated content
 *   Copyright (C) 2011 Glenn Engstrand
 *
 *   This program is free software: you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation, either version 3 of the License, or
 *   (at your option) any later version.
 *
 *   This program is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 * 
 *   You should have received a copy of the GNU General Public License
 *   along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
package com.dynamicalsoftware.news;

import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.FileSystemXmlApplicationContext;

/**
 * facade for both RabbitMQ and Cassandra
 * @author glenn
 *
 */
public class NewsWidgetActivity extends Thread {

	private static final long serialVersionUID = 2L;
	private static final String CHANNEL_NAME = "news";
	private static final String SHUTDOWN_CMD = "done";
	private static final Logger log = Logger.getLogger(NewsWidgetActivity.class.getCanonicalName());
		
	private NewsPublisher publisher = null;
	
	@Override
	public void run() {
		super.run();
		Connection conn = null;
		Channel chan = null;
		try {
			conn = getConnection();
			chan = getChannel(conn);
		    QueueingConsumer consumer = new QueueingConsumer(chan);
		    chan.basicConsume(CHANNEL_NAME, true, consumer);
		    boolean more = true;
		    while (more) {
		    	QueueingConsumer.Delivery delivery = null;
				try {
					delivery = consumer.nextDelivery();
					String content = new String(delivery.getBody());
					if (content.equals(SHUTDOWN_CMD)) {
						more = false;
					} else {
						String[] msg = content.split("|");
						if (msg.length == 4) {
							getPublisher().publish(msg[0], msg[1], msg[2], msg[3]);
						} else {
							log.log(Level.WARNING, "Malformed request: {0}", content);
						}
					}
				} catch (ShutdownSignalException e) {
					more = false;
				} catch (InterruptedException e) {
					more = false;
				}
		    }
		} catch (IOException e) {
			log.log(Level.SEVERE, "Error retrieving from RabbitMQ.\n", e);
		} finally {
			if (chan != null) {
				try {
					chan.close();
				} catch (IOException e) {
					log.log(Level.WARNING, "Cannot close RabbitMQ channel.\n", e);
				}
			}
			if (conn != null) {
				try {
					conn.close();
				} catch (IOException e) {
					log.log(Level.WARNING, "Cannot close RabbitMQ connection.\n", e);
				}
			}
		}
	}
	
	/**
	 * initiate shutdown for RabbitMQ consumer thread
	 */
	public void shutdown() {
		publish(SHUTDOWN_CMD, "", "", "");
	}
	
	/**
	 * send a display item activity notice 
	 * @param key - used by cassandra to uniquely identify this item
	 * @param title is the title of the item being displayed
	 * @param link is the URL of the item being displayed
	 * @param description is the description of the item being displayed
	 */
	public void publish(String key, String title, String link, String description) {
		StringBuffer sb = new StringBuffer();
		sb.append(key);
		sb.append("|");
		sb.append(title);
		sb.append("|");
		sb.append(link);
		sb.append("|");
		sb.append(description);
		Connection conn = null;
		Channel chan = null;
		try {
			conn = getConnection();
			chan = getChannel(conn);
			chan.basicPublish("", CHANNEL_NAME, null, sb.toString().getBytes());
		} catch (IOException e) {
			log.log(Level.SEVERE, "Cannot publish to RabbitMQ.\n", e);
			getPublisher().publish(key, title, link, description);
		} finally {
			if (chan != null) {
				try {
					chan.close();
				} catch (IOException e) {
					log.log(Level.WARNING, "Cannot close RabbitMQ channel.\n", e);
				}
			}
			if (conn != null) {
				try {
					conn.close();
				} catch (IOException e) {
					log.log(Level.WARNING, "Cannot close RabbitMQ connection.\n", e);
				}
			}
		}
	}
	
	/**
	 * get a connection to rabbitmq
	 * @return rabbit mq connection
	 * @throws IOException
	 */
	private Connection getConnection() throws IOException {
	    ConnectionFactory factory = new ConnectionFactory();
	    factory.setHost("localhost");
	    return factory.newConnection();
	}
	
	/**
	 * get the news channel from the rabbitmq connection
	 * @param conn rabbitmq connection
	 * @return channel
	 * @throws IOException
	 */
	private Channel getChannel(Connection conn) throws IOException {
	    Channel retVal = conn.createChannel();
	    retVal.queueDeclare(CHANNEL_NAME, false, false, false, null);
	    return retVal;
	}
	
	/**
	 * facade for accessing the data store
	 * @return backing store specific publisher 
	 */
	private NewsPublisher getPublisher() {
		if (publisher == null) {
			ApplicationContext context = new FileSystemXmlApplicationContext("webapps/NewsWidget/WEB-INF/newsWidget.xml");
			publisher = (NewsPublisher)context.getBean("news");
		}
		return publisher;
	}
}
